package com.atguigu.flink.window;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;

/**
 * Created by Smexy on 2023/4/8
 *
 *  求每种传感器，每5条数据中vc最大的前top3的信息。
 */
public class Demo9_Process
{
    public static void main(String[] args) throws Exception {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

                env
                   .socketTextStream("hadoop102", 8888)
                   .map(new WaterSensorMapFunction())
                   .keyBy(WaterSensor::getId)
                   .countWindow(5)
                   //.evictor()  回收器。 用在滑动窗口。决定窗口中进入的元素，在触发运算时，是否要删除一部分，再运算
                   //.trigger()  触发器，调用窗口的api时，它都有默认的触发器。滚动窗口，每slide触发一次运算。如果希望修改窗口触发运算的时机，自己传入trigger
                   .process(new ProcessWindowFunction<WaterSensor, String, String, GlobalWindow>()
                   {
                       //非滚动聚合，等窗口触发运算时，只执行一次
                       @Override
                       public void process(String key, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {

                           //对集合运算，一定使用Stream方便
                           List<WaterSensor> top3 = StreamSupport.stream(elements.spliterator(), true)
                                                                    .sorted((w1, w2) -> -w1.getVc().compareTo(w2.getVc()))
                                                                    .limit(3)
                                                                    .collect(Collectors.toList());

                           out.collect(key + ":" +top3);


                       }
                   })
                   .print();


            env.execute();


    }
}
